package com.lagou.java;

import com.sun.jndi.cosnaming.CNCtx;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.timeout.ReadTimeoutHandler;
import io.netty.handler.timeout.WriteTimeoutHandler;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;

import javax.xml.soap.Node;
import java.lang.reflect.Proxy;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.zip.GZIPOutputStream;

/**
 * @Classname RpcConsumer
 * @Description TODO
 * @Date 2020/7/22 13:11
 * @Created by pc
 */
public class RpcConsumer {

    private static ExecutorService executor =
            Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    // 标记发往哪个服务端
    private static int sendIndex = 0;


    private static Map<String, NettyClientInfo> clientData = new ConcurrentHashMap<>();

    // zookeeper 客户端
    private static ZkClient zkClient= null;

    // netty 客户端根节点
    public static final String clientBasePath = "/netty_client";

    // netty 服务器根节点
    public static final String serverBasePath = "/netty_provider";

    // 初始化 zookeeper
    public static void initZkClient() throws Exception {
        // 创建连接
        zkClient = zkClient == null ? new ZkClient("127.0.0.1:2181") : zkClient;
        System.out.println("Zookeeper session established. address is :" + "127.0.0.1:2181");
        // 创建父节点
        if (!zkClient.exists(clientBasePath)) {
            zkClient.createPersistent(clientBasePath, true);
            System.out.println("==>> 父节点：" + clientBasePath + ", 创建成功！");
        }

        // 获取服务端所有子节点
        List<String> children = zkClient.getChildren(serverBasePath);
//        serverAddress = serverAddress == null ? new ConcurrentHashMap<>(): serverAddress;
        children.forEach(path -> {
            String address = zkClient.readData(serverBasePath +"/" + path);
            NettyClientInfo nettyClientInfo = new NettyClientInfo();
            nettyClientInfo.setLocalAddress(address);
            clientData.put(path, nettyClientInfo);
        });
        System.out.println("===================================");
        children.forEach(System.out::println);
        System.out.println("===================================");

        // 注册监听事件
        zkClient.subscribeChildChanges(serverBasePath, new IZkChildListener() {
            public synchronized void handleChildChange(String parentPath,
                                                       List<String> currentChilds) throws Exception {
                currentChilds = zkClient.getChildren(serverBasePath);
//                System.out.println("servercache " + serverAddress);
                System.out.println("current : " + currentChilds);
                for (String node : currentChilds) {
                    if (!clientData.keySet().contains(node)) {
                        String address = zkClient.readData(parentPath + "/" + node);
                        System.out.println(node + " > 新增服务端节点, 节点信息为: " + address);
                        NettyClientInfo info = clientData.get(node);
                        if (!info.getRemoteAdress().contains(address)) {
                            initClient(info, address);
                        }
                        registerClient(address, info);
                    }
                }
                for (String node : clientData.keySet()) {
                    if (!currentChilds.contains(node)) {
                        System.out.println(node + " > 被删除");
                        String address = zkClient.readData(parentPath + "/" + node);
                        NettyClientInfo clientInfo = clientData.get(node);
                        // 断开和服务端的连接
                        clientInfo.getChannelFuture().get(address).channel().close();
                    }
                }

                System.out.println("---------------");

                // 打印缓存数据
                System.out.println(parentPath + " 's child changed currentChilds:" + currentChilds);
            }
        });

    }


    /**
     * 创建一个代理对象
     */
    public synchronized Object createProxy(final Class<?> serviceClass, NettyClientInfo clientInfo) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class<?>[]{serviceClass}, (proxy, method, args) -> {
//                    if (clientInfo == null) {
//                        initClient(clientData.get(node));
//                    }
                    // 设置参数
                    RpcRequest request = new RpcRequest();
                    request.setRequestId(UUID.randomUUID().toString());
                    request.setClassName(serviceClass.getName());
                    request.setMethodName(method.getName());
                    request.setParameterTypes(method.getParameterTypes());
                    request.setParameters(args);
                    // 轮询发送消息
                    List<String> addresses = clientInfo.getRemoteAdress();
                    UserClientHandler client = (UserClientHandler) clientInfo.getHandler();
                    Channel channel = clientInfo.getChannelFuture().get(sendIndex % addresses.size()).channel();
                    // 初始化发送索引
                    sendIndex = (sendIndex + 1) % addresses.size();
                    // 发送消息
//                    return executor.submit(client).get();
                    return null;
                });
    }

    // 初始化客户端
    private static synchronized void initClient(NettyClientInfo clientInfo, String address) throws Exception {
        if (zkClient == null) {
            initZkClient();
//            addresses = serverAddress.values().toArray(new String[serverAddress.size()]);
        }
        // 初始化客户端
        UserClientHandler clientHandler = (UserClientHandler) clientInfo.getHandler();
        clientHandler = clientHandler == null ? new UserClientHandler() : clientHandler;
        clientInfo.setHandler(clientHandler);

        try {
            initBootstrap(clientHandler, clientInfo);
            registerClient(address, clientInfo);
        } finally {
            if (clientInfo.getGroup() != null) {
                clientInfo.getGroup().shutdownGracefully();
            }
        }
    }

    // 初始化 netty 启动器
    public static void initBootstrap(UserClientHandler clientHandler, NettyClientInfo clientInfo) {
        EventLoopGroup group = null;
        group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();

        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ChannelPipeline pipeline = ch.pipeline();
//                        pipeline.addLast(new StringEncoder());
                        pipeline.addLast(new StringDecoder());
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(clientHandler);
//                            //实现5秒钟，如果两端，如果数据读取，直接断开连接
//                            pipeline.addLast(new ReadTimeoutHandler(5));
//                            //实现5秒钟，如果两端，如果数据写入，直接断开连接
//                            pipeline.addLast(new WriteTimeoutHandler(5));
                    }
                });
        clientInfo.setGroup(group);
        clientInfo.setBootstrap(bootstrap);

    }

    // 向 zookeeper 注册消费者节点 地址
    public static void registerClient (String address, NettyClientInfo clientInfo) {
        ChannelFuture channelFuture = null;
        try {
            String[] addressInfo = address.split(":");
            channelFuture = clientInfo.getBootstrap().connect(addressInfo[0], Integer.parseInt(addressInfo[1])).sync();
            Map<String, ChannelFuture> futureMap = clientInfo.getChannelFuture();
            futureMap.put(address, channelFuture);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // 将当前服务节点注册到 zookeeper
            InetSocketAddress socketAddress = (InetSocketAddress)channelFuture.channel().localAddress();
            String path = zkClient.createEphemeralSequential(clientBasePath + "/customer", socketAddress.getHostString()+":" + socketAddress.getPort());
            System.out.println("==> path : " + path + "创建 成功！");
        }
    }


}
